package com.raft.core.rpc.nio;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;

import com.alibaba.com.caucho.hessian.io.HessianInput;
import com.raft.core.node.NodeId;
import com.raft.core.rpc.message.AppendEntriesResult;
import com.raft.core.rpc.message.AppendEntriesRpc;
import com.raft.core.rpc.message.InstallSnapshotResult;
import com.raft.core.rpc.message.InstallSnapshotRpc;
import com.raft.core.rpc.message.MessageConstants;
import com.raft.core.rpc.message.RequestVoteResult;
import com.raft.core.rpc.message.RequestVoteRpc;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

/**解析请求的类，把对象转换为二进制数据<br/>
 * int		int		var<br/>
 * type		len		payload
 * @author zhangsulei
 *
 */
public class Decoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    	// 预读8个字节(消息类型和负载长度)
        int availableBytes = in.readableBytes();
        if (availableBytes < 8) return;

        // 记录当前的起始位置
        in.markReaderIndex();
        int messageType = in.readInt();
        int payloadLength = in.readInt();
        // 消息尚未完全可读(半包状态)
        if (in.readableBytes() < payloadLength) {
        	// 回到起始位置
            in.resetReaderIndex();
            return;
        }

        // 消息可读
        byte[] payload = new byte[payloadLength];
        in.readBytes(payload);
        // 根据消息类型
        switch (messageType) {
            case MessageConstants.MSG_TYPE_NODE_ID:
                out.add(new NodeId(new String(payload)));
                break;
            case MessageConstants.MSG_TYPE_REQUEST_VOTE_RPC:
                RequestVoteRpc rpc = decode(payload, RequestVoteRpc.class);
                out.add(rpc);
                break;
            case MessageConstants.MSG_TYPE_REQUEST_VOTE_RESULT:
                RequestVoteResult protoRVResult = decode(payload, RequestVoteResult.class);
                out.add(protoRVResult);
                break;
            case MessageConstants.MSG_TYPE_APPEND_ENTRIES_RPC:
                AppendEntriesRpc aeRpc = decode(payload, AppendEntriesRpc.class);
                out.add(aeRpc);
                break;
            case MessageConstants.MSG_TYPE_APPEND_ENTRIES_RESULT:
                AppendEntriesResult appendEntriesResult = decode(payload, AppendEntriesResult.class);
                out.add(appendEntriesResult);
                break;
            case MessageConstants.MSG_TYPE_INSTALL_SNAPSHOT_PRC:
                InstallSnapshotRpc isRpc = decode(payload, InstallSnapshotRpc.class);
                out.add(isRpc);
                break;
            case MessageConstants.MSG_TYPE_INSTALL_SNAPSHOT_RESULT:
                InstallSnapshotResult installSnapshotResult = decode(payload, InstallSnapshotResult.class);
                out.add(installSnapshotResult);
                break;
        }
    }

    private <T extends Serializable> T decode(byte[] bytes, Class<T> clazz) {
        if (bytes == null) {
            return null;
        }
        // 1、将字节数组转换成字节输入流
        ByteArrayInputStream bis = new ByteArrayInputStream(bytes);

        // step 1. 定义外部序列化工厂
        //ExtSerializerFactory extSerializerFactory = new ExtSerializerFactory();
        //extSerializerFactory.addSerializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisSerializer());
        //extSerializerFactory.addDeserializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisDeserializer());
        // step 2. 序列化工厂
        //SerializerFactory serializerFactory = new SerializerFactory();
        //serializerFactory.addFactory(extSerializerFactory);
        HessianInput hessianInput = new HessianInput(bis);
        //hessianInput.setSerializerFactory(serializerFactory);
        Object object = null;

        try {
            object = hessianInput.readObject();
        } catch (IOException e) {
        	e.printStackTrace();
        } catch (Exception e) {
        	e.printStackTrace();
        }

        return (T) object;
    }

}
